package com.fsl.springbootstreamkafka.original;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

/**
 * @author dell
 * kafka原生API调用----producer
 */
public class KafkaProducerTest {

    public static void main(String[] args) throws Exception {
        //kafka配置文件
        Properties properties = new Properties();
        //bootstrap.servers=localhost:9092
        properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost:9092");
        properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());

        //建立kafka producer
        KafkaProducer producer = new KafkaProducer<String, String>(properties);

        String topic = "mytopic";

        for(int i=0;i<10;i++){
            /* 发送消息 */
            producer.send(new ProducerRecord(topic,String.valueOf(i),"hello jullie" + i),new Callback() {
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    System.out.println("message send to ----->" + recordMetadata.partition() + " ,offset----->" + recordMetadata.offset());
                }
            });
        }

        producer.send(new ProducerRecord("mytopic","1","hello world"),(recordMetadata,e)-> {
            System.out.println("message send to ----->" + recordMetadata.partition() + " ,offset----->" + recordMetadata.offset());
        });
        System.in.read();
    }
}
